import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/10/11 09:11
 */
public class Join_2 extends BaseSQLApp {
    public static void main(String[] args) {
        new Join_2().init(4000, 1, "Join_1");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table a(" +
                            "id int, " +
                            "name string)" + SQLUtil.getKafkaSource("a", "Join_1"));
        
        tEnv.executeSql("create table b(" +
                            "id int, " +
                            "age int)" + SQLUtil.getKafkaSource("b", "Join_1"));
        
        
        // 给状态设置一个 ttl
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        tEnv
            .sqlQuery("select " +
                          "a.id, " +
                          "name, " +
                          "age " +
                          "from a " +
                          "left outer join b on a.id=b.id")
            .execute()
            .print();
    }
}
/*
join:
    默认情况下, 会把数据永远存在状态中
    
left join
    如果右表先来, 则正常的每条数据都是新增
    
    如果左表先来, 会有一条结果是 左表有数据, 右表的字段用 null 补齐, 现在如果右表来了,在先删除,在再新增.以后再来都是新增
    
    对左表来说, ttl 表示的是 idle 的时间之后会清空数据
 */